package com.google.appengine.tools.mapreduce.inputs;

import java.io.IOException;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;

import com.google.appengine.api.blobstore.BlobKey;
import com.google.appengine.api.blobstore.BlobstoreInputStream;
import com.google.appengine.tools.mapreduce.InputReader;
import com.google.appengine.tools.mapreduce.KeyValue;
import com.google.appengine.tools.mapreduce.inputs.InputStreamIterator.OffsetRecordPair;
import com.google.common.base.Preconditions;
import com.google.common.io.CountingInputStream;

/**
 * BlobstoreZipLineInputReader reads a zip entry from a zip file from blobstore one line at a time.
 * <p>
 * Use of the <code>com.google</code> package was due to some dependencies that had package visibility
 * (e.g. InputStreamIterator).
 */
class BlobstoreZipInputReader extends InputReader<KeyValue<String, byte[]>> {

	private static final long serialVersionUID = 4206774935224030898L;

	private static final Logger log = Logger.getLogger(BlobstoreZipInputReader.class.getName());

	/* VisibleForTesting */long startOffset;
	/* VisibleForTesting */long endOffset;
	private String blobKey;
	private byte terminator;
	private long offset = 0L;
	/* VisibleForTesting */long zipEntryUncompressedSize;
	/* VisibleForTesting */String zipEntryName;
	private transient CountingInputStream countingInput;
	private transient Iterator<OffsetRecordPair> recordIterator;

	public BlobstoreZipInputReader(String blobKey, long startOffset,
			long endOffset, byte terminator, ZipEntry zipEntry) {
		this.blobKey = blobKey;
		this.startOffset = startOffset;
		this.endOffset = endOffset;
		this.zipEntryName = zipEntry.getName();
		// this.zipEntryCompressedSize = zipEntry.getCompressedSize();
		this.zipEntryUncompressedSize = zipEntry.getSize();
		this.terminator = terminator;
	}

	private void checkInitialized() {
		Preconditions.checkState(recordIterator != null, "%s: Not initialized",
				this);
	}

	@Override
	public KeyValue<String, byte[]> next() {
		checkInitialized();
		if (!recordIterator.hasNext()) {
			throw new NoSuchElementException();
		}
		
		// TODO(ldee): similar to BlobstoreInputReader, simplify by removing OffsetRecordPair
		return KeyValue.<String, byte[]>of(this.zipEntryName, recordIterator.next().getRecord());
	}

	@Override
	public Double getProgress() {
		checkInitialized();
		// TODO(ldee): Would be good to have access to ZipInputStream's
		// java.util.zip.Inflater object and get the number of bytes read
		// via #getBytesRead(). This would provide more accurate progress.
		if (zipEntryUncompressedSize == 0) {
			return 1.0;
		} else {
			double currentOffset = offset + countingInput.getCount();
			return Math.min(1.0, currentOffset / zipEntryUncompressedSize);
		}
	}

	@Override
	public void beginSlice() throws IOException {
		Preconditions.checkState(recordIterator == null,
				"%s: Already initialized: %s", this, recordIterator);
		try {
			/*
			 * This code works on local devserver, but not on app engine (sigh).
			 * 
			ZipInputStream zipInput = new ZipInputStream(
					new BlobstoreInputStream(new BlobKey(blobKey), startOffset));
			zipInput.getNextEntry();
			if (offset > 0L) {
				// skip until offset bytes in the current zip entry
				log.info("Skipping to " + offset + "...");
				zipInput.skip(offset);
				log.info("Skipped to " + offset);
			}
			*/
			ZipInputStream zipInput = new ZipInputStream(
					new BlobstoreInputStream(new BlobKey(blobKey)));
			ZipEntry zipEntry;
			while ((zipEntry = zipInput.getNextEntry()) != null) {
				if (zipEntry.getName().equals(this.zipEntryName)) {
					break;
				}
			}
			countingInput = new CountingInputStream(zipInput);
			recordIterator = new InputStreamIterator(
					countingInput, zipEntryUncompressedSize,
					offset == 0L, terminator);
		} catch (IOException e) {
			log.log(Level.WARNING, "Error beginning slice @" + offset, e);
			throw e;
		}
	}

	@Override
	public void endSlice() throws IOException {
		checkInitialized();
		offset += countingInput.getCount();
		countingInput.close();
		// Un-initialize to make checkInitialized() effective.
		countingInput = null;
		recordIterator = null;
	}

}
